package net.bwie.jtp.dwd.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.guanjuntao.util.JdbcUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class JtpLogClickHouse {
    private static TableEnvironment getTableEnv() {
        // TODO: 2025/5/22 创建执行环境
        EnvironmentSettings build = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useAnyPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(build);
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone","Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism","1");
        configuration.setString("table.exec.state.ttl","5 s");
        return tabEnv ;
    }

    public static void savePageLog(DataStream<String> stream) {
        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        // TODO: 2025/5/22 解析字段
                        JSONObject object = JSON.parseObject(value);
                        JSONObject commonObject = object.getJSONObject("common");
                        JSONObject pageObject = object.getJSONObject("page");
                        // TODO: 2025/5/22 提取字段
                        JSONObject jsonObject = new JSONObject() ;
                        jsonObject.put("mid", commonObject.getString("mid")) ;
                        jsonObject.put("area_code", commonObject.getString("ar")) ;
                        jsonObject.put("brand", commonObject.getString("ba")) ;
                        jsonObject.put("channel", commonObject.getString("ch")) ;
                        jsonObject.put("is_new", commonObject.getString("is_new")) ;
                        jsonObject.put("model", commonObject.getString("md")) ;
                        jsonObject.put("os", commonObject.getString("os")) ;
                        jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString()) ;
                        jsonObject.put("version_code", commonObject.getString("vc")) ;
                        jsonObject.put("page_id", pageObject.getOrDefault("page_id", "")) ;
                        jsonObject.put("last_page_id", pageObject.getOrDefault("last_page_id", "")) ;
                        jsonObject.put("during_time", pageObject.getLong("during_time")) ;
                        jsonObject.put("sourceType", pageObject.getOrDefault("source_type", "")) ;
                        jsonObject.put("item_type", pageObject.getOrDefault("item_type", "")) ;
                        jsonObject.put("item", pageObject.getOrDefault("item", "")) ;
                        jsonObject.put("ts", object.getLong("ts")) ;
                        // 3返回JSON字符串
                        return jsonObject.toJSONString();

                    }
                });
        String[] columns =new String[] {
                "mid", "area_code", "brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "page_id", "last_page_id", "during_time", "sourceType", "item_type", "item", "ts"
        };
        // TODO: 2025/5/22 插入数据
        String upsertSql = "" +
                "INSERT INTO jtp_app_log.dwd_traffic_page_log(mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "                                             page_id, last_page_id, during_time, sourceType, item_type, item,\n" +
                "                                             ts)\n" +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
        // TODO: 2025/5/22 插入数据
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream, columns, upsertSql);
    }




    /**
     * 实时将APP启动日志数据，写入ClickHouse表中
     */
    public static void saveStartLog(DataStream<String> stream) {
        // 1数据转换
        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        //TODO: 2025/5/22 解析JSON
                        JSONObject object = JSON.parseObject(value);
                        JSONObject commonObject = object.getJSONObject("common");
                        JSONObject startObject = object.getJSONObject("start");

                        //TODO: 2025/5/22 提前字段值
                        JSONObject jsonObject = new JSONObject() ;
                        jsonObject.put("mid", commonObject.getString("mid")) ;
                        jsonObject.put("area_code", commonObject.getString("ar")) ;
                        jsonObject.put("brand", commonObject.getString("ba")) ;
                        jsonObject.put("channel", commonObject.getString("ch")) ;
                        jsonObject.put("is_new", commonObject.getString("is_new")) ;
                        jsonObject.put("model", commonObject.getString("md")) ;
                        jsonObject.put("os", commonObject.getString("os")) ;
                        jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString()) ;
                        jsonObject.put("version_code", commonObject.getString("vc")) ;

                        jsonObject.put("entry", startObject.getString("entry")) ;
                        jsonObject.put("loading_time", startObject.getLong("loading_time")) ;
                        jsonObject.put("open_ad_id", startObject.getLong("open_ad_id")) ;
                        jsonObject.put("open_ad_ms", startObject.getLong("open_ad_ms")) ;
                        jsonObject.put("open_ad_skip_ms", startObject.getLong("open_ad_skip_ms")) ;

                        jsonObject.put("ts", object.getLong("ts")) ;
                        //TODO: 2025/5/22 返回JSON字符串
                        return jsonObject.toJSONString();
                    }
                }
        );

        // TODO: 2025/5/22 字段名称
        String[] columns = new String[]{
                "mid", "area_code","brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "entry", "loading_time", "open_ad_id", "open_ad_ms", "open_ad_skip_ms",  "ts"
        };
        // TODO: 2025/5/22  插入语句
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_app_start_log(\n" +
                "    mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "    entry, loading_time, open_ad_id, open_ad_ms, open_ad_skip_ms,\n" +
                "    ts\n" +
                ") VALUES (\n" +
                "     ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?\n" +
                " )" ;
        // TODO: 2025/5/22插入数据
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream,columns, upsertSql);
    }


    public static void saveErrLog(DataStream<String> stream){

        SingleOutputStreamOperator<String> jsonStream = stream.map(
                new RichMapFunction<String, String>() {
                    @Override
                    public String map(String value) throws Exception {
                        JSONObject Object = JSON.parseObject(value);
                        JSONObject commonObject = Object.getJSONObject("common");
                        JSONObject errorObject = Object.getJSONObject("err");

                        JSONObject jsonObject = new JSONObject();
                        // todo common
                        jsonObject.put("mid", commonObject.getString("mid"));
                        jsonObject.put("area_code", commonObject.getString("ar"));
                        jsonObject.put("brand", commonObject.getString("ba"));
                        jsonObject.put("channel", commonObject.getString("ch"));
                        jsonObject.put("is_new", commonObject.getString("is_new"));
                        jsonObject.put("model", commonObject.getString("md"));
                        jsonObject.put("os", commonObject.getString("os"));
                        jsonObject.put("uid", commonObject.getOrDefault("uid", "").toString());
                        jsonObject.put("version_code", commonObject.getString("vc"));
                        // todo err
                        jsonObject.put("msg", errorObject.getString("msg").replace("\\", ""));
                        jsonObject.put("error_code", errorObject.getOrDefault("error_code", ""));

                        //todo page
                        if (null != Object.get("page")) {
                            JSONObject pageObject = Object.getJSONObject("page");
                            jsonObject.put("page_id", pageObject.getOrDefault("page_id", ""));
                            jsonObject.put("last_page_id", pageObject.getOrDefault("last_page_id", ""));
                            jsonObject.put("during_time", pageObject.getLong("during_time"));
                            jsonObject.put("sourceType", pageObject.getOrDefault("sourceType", ""));
                            jsonObject.put("item_type", pageObject.getOrDefault("item_type", ""));
                            jsonObject.put("item", pageObject.getOrDefault("item", ""));
                        } else {
                            jsonObject.put("page_id", "");
                            jsonObject.put("last_page_id", "");
                            jsonObject.put("during_time", "");
                            jsonObject.put("sourceType", "");
                            jsonObject.put("item_type", "");
                            jsonObject.put("item", "");
                        }

                        // todo start
                        if (null != Object.get("start")) {
                            JSONObject startObject = Object.getJSONObject("start");
                            jsonObject.put("entry", startObject.getString("entry"));
                            jsonObject.put("loading_time", startObject.getLong("loading_time"));
                            jsonObject.put("open_ad_id", startObject.getLong("open_ad_id"));
                            jsonObject.put("open_ad_ms", startObject.getLong("open_ad_ms"));
                            jsonObject.put("open_ad_skip_ms", startObject.getLong("open_ad_skip_ms"));
                        } else {
                            jsonObject.put("entry", "");
                            jsonObject.put("loading_time", "");
                            jsonObject.put("open_ad_id", "");
                            jsonObject.put("open_ad_ms", "");
                            jsonObject.put("open_ad_skip_ms", "");
                        }
                        // todo displays
//                        jsonObject.put("display_type",displaysObject.getString("display_type"));
                        if (null != Object.get("displays")) {
                            JSONArray displaysObject = Object.getJSONArray("displays");
                            jsonObject.put("displays", displaysObject.toJSONString());
                        } else {
                            jsonObject.put("displays", "");
                        }
                        // todo  actions
                        if (null != Object.get("actions")) {
                            JSONArray actionsObject = Object.getJSONArray("actions");
                            jsonObject.put("actions", actionsObject.toJSONString());
                        } else {
                            jsonObject.put("actions", "");
                        }
                        // 时间戳
                        jsonObject.put("ts", Object.getLong("ts"));
                        return jsonObject.toJSONString();
                    }
                }
        );
        String[] columns = new String[]{
                "mid", "area_code", "brand", "channel", "is_new", "model", "os", "uid", "version_code",
                "msg", "error_code", "page_id", "last_page_id", "during_time", "sourceType",
                "item_type", "item", "entry", "loading_time", "open_ad_id", "open_ad_ms",
                "open_ad_skip_ms", "displays", "actions", "ts"
        };
        String upsertSql = "INSERT INTO jtp_app_log.dwd_traffic_app_error_log (" +
                "mid, area_code, brand, channel, is_new, model, os, uid, version_code,\n" +
                "msg, error_code, page_id, last_page_id, during_time, sourceType,\n" +
                "item_type, item, entry, loading_time, open_ad_id, open_ad_ms,\n" +
                "open_ad_skip_ms, displays, actions, ts)\n" +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
        JdbcUtil.sinkToClickhouseUpsertJson(jsonStream, columns, upsertSql);
    }


}
